import time

import pandas as pd

from datetime import datetime
from math import fabs
import threading
from collections import defaultdict
from queue import Queue
from threading import Thread
from time import sleep
# from tkinter.messagebox import NO
from typing import Any, Callable, List
import json
import os
import yaml
import marshal
import redis
from redis import StrictRedis

INSTRUMENT = "inst."
FUTURE = "future."
OPTION = "option."
ORDER = "order."
PARKORDER = "park_order."
TRADE = "trade."
ACCOUNT = "acc."
POSITION = "position."
POSITIONDETAIL = "position_detail."
STRATEGY = "strategy."
STRASIGNAL = "eStraSignal."
TICK = "Tick."  # 后加 instrument id, publish 时的 channel
BAR = "Bar."

SUBSCRIBEINSTRUMENT = "Subscribe."
REQSUBINST = "ReqSubInst."
REQORDERINSERT = "ReqOrderInsert."
REQUNSUBINST = "ReqUnSubInst."
RQACC = "ReqQryAccount."
RQTRADE = "ReqQryTrade."
RQORDER = "ReqQryOrder."
RQPOSITION = "ReqQryPosition."
RQPOSITIONDETAIL = "ReqQryPositionDetail."

# # PUBLISH<=>SUBSCRIBE 用作channel
CTPWORK = "CTPWORK"  # msg:StartQuote,StopQuote,StartTrade,StopTrade
EVENT_MD_CONNECT_STATUS = "QuoteConnected"  # true or false  1 or 0
EVENT_MD_LOGIN_STATUS = "QuoteLogin"  # true or false
EVENT_TD_CONNECT_STATUS = "TradeConnected"
EVENT_TD_LOGIN_STATUS = "TradeLogin"
# # ctp redis 中数据库
COMMONDB = 0  # 通用数据库
QUOTEDB = 1  # 市场行情数据库
TRADEDB = 2  # 交易数据库
ACCOUNTDB = 10  # 账户数据库
STRATEGYDB = 3  # 策略数据库

EVENT_RQ = "rq_"  # 查询请求，可与其它联合使用：EVENT_RQ+EVENT_ORDER 生成 rq_eOrder.
EVENT_TIMER = "eTimer"  # noqa
EVENT_TICK = "eTick."
EVENT_TRADE = "eTrade."
EVENT_ORDER = "eOrder."
EVENT_PARK_ORDER = "eParkOrder."
EVENT_POSITION = "ePosition."
EVENT_ACCOUNT = "eAccount."
EVENT_CONTRACT = "eContract."
EVENT_LOG = "eLog"
EVENT_POSITION_DETAIL = "ePositionDetail."
EVENT_MD_LOGIN = "eMDLogin."
EVENT_TD_LOGIN = "eTDLogin."
EVENT_MD_CONNECTED = "eMDConnected."
EVENT_TD_CONNECTED = "eTDConnected."
EVENT_INSTRUMENT = "eInstrument."
EVENT_INSTRUMENT_OK = "eInstrumentOK."
EVENT_POSITION_DETAIL_OK = "ePositionDetailOK."
EVENT_POSITION_OK = "ePositionOK."
EVENT_INSTRUMENT_STATUS = "eInstrumentStatus."
EVENT_ACCOUNT_OK = "eAccountOK."
EVENT_ORDER_OK = "eOrderOK."
EVENT_FUTURE = "eFuture."
EVENT_STRA = "eStrategy."
EVENT_STRATEGY = "eStrategy."
EVENT_STRA_SIGNAL = "eStraSignal."
EVENT_STOP_ORDER = "eStopOrder."
EVENT_SUBSCRIBE = "eSubscribe."
EVENT_UNSUBSCRIBE = "eUnSubscribe."

EVENT_REQ_SUBSCRIBE = "eReqSubscribe." # 后跟 订阅的类型或者具体标的
EVENT_REQ_SUBSCRIBE_TICK = "tick."
EVENT_REQ_SUBSCRIBE_BAR = "md."
EVENT_ON_TICK="eOnTick."
EVENT_ON_BAR="eOnBar."

class Event:
    """
    Event object consists of a type string which is used
    by event engine for distributing event, and a data
    object which contains the real data.
    """

    def __init__(self, event_type: str, data: Any = None):
        """"""
        self.event_type: str = event_type
        self.data: Any = data


# Defines handler function to be used in event engine.
HandlerType = Callable[[Event], None]


class RedisEventEngine:
    """
    Event engine distributes event object based on its type
    to those handlers registered.

    It also generates timer event by every interval seconds,
    which can be used for timing purpose.
    """

    def __init__(self, interval: int = 1, rds: StrictRedis = None):
        """
        Timer event is generated every 1 second by default, if
        interval not specified.
        """
        if rds is None:
            rds = rds
            # pool = redis.ConnectionPool(host=cfg.settings['redis']['host'], port=cfg.settings['redis']['port'], db=0,
            #                             decode_responses=True)
            # rds = redis.StrictRedis(connection_pool=pool)
        self._interval: int = interval
        self._rds: StrictRedis = rds

        self._active: bool = False
        self._thread: Thread = Thread(target=self._run)
        self._timer: Thread = Thread(target=self._run_timer)
        self._handlers: defaultdict = defaultdict(list)
        self._tick_handlers: defaultdict = defaultdict(list)
        self._bar_handlers: defaultdict = defaultdict(list)
        self._general_handlers: List = []
        self._listened_channels = []
        self._listener = []
        self.ps = self._rds.pubsub()  # redis.pubsub()
        # iid:[eid,pid,need_subscribed,subscribed,main_second(0,1,2)]
        self.iids = {}

        self._queue = Queue()
        self._channels_queue = Queue()

        self._req_thread: Thread = Thread(target=self._on_reqsubscribe)

    def _run(self) -> None:
        """
        Get event from queue and then process it.
        """

        tc = 0

        while self._active:

            # print(self._channels)
            # for itm in [k for k in self._channels.keys() if self._channels[k] == 0]:
            #     print('\n', itm, ":", tc)
            #     self.ps.psubscribe(itm)
            # self._channels[itm] = 1
            try:
                itm = self._channels_queue.get(block=True, timeout=1)
                if isinstance(itm, Event):
                    if itm.event_type == EVENT_REQ_SUBSCRIBE:
                        print(itm.data)
                        threading.Thread(target=self._listen, ).start()
            except:
                pass
                tc += 1

    def _listen(self):
        for data in self.ps.listen():
            if data['type'] in ['message', 'pmessage']:
                ed = data['data']

                if isinstance(ed, str):
                    ed = json.loads(ed)
                threading.Thread(target=self._process, args=(
                    Event(data['channel'], ed),)).start()

    def _tick_listen(self,ps=None):
        """tick 订阅后调用"""
        if ps is None:
            ps=self.ps
        for data in ps.listen():
            if data['type'] in ['message', 'pmessage']:
                ed = data['data']
                ed = json.loads(ed)
                iid = data['channel'].split('.')[1]
                threading.Thread(target=self._process_tick, args=(
                    iid, ed,)).start()

    def _bar_listen(self,ps=None):
        """ bar 订阅后调用"""
        if ps is None:
            ps=self.ps
        for data in ps.listen():
            if data['type'] in ['message', 'pmessage']:
                ed = data['data']
                ed = json.loads(ed)
                iid = data['channel'].split('.')[1]
                threading.Thread(target=self._process_bar, args=(
                    iid, ed,)).start()

    def _process(self, event: Event) -> None:
        """
        First ditribute event to those handlers registered listening
        to this type.

        Then distrubute event to those general handlers which listens
        to all types.
        """
        if event.event_type in self._handlers:
            [handler(event) for handler in self._handlers[event.event_type]]

        if self._general_handlers:
            [handler(event) for handler in self._general_handlers]

    def _process_tick(self, iid, tick) -> None:
        """
        收到订阅的 tick 后调用
        iid: instrument id
        tick: tick

        """
        # print(tick)
        [handler(tick) for handler in self._tick_handlers[iid]]

    def _process_bar(self, iid, bar) -> None:
        """
        收到订阅的 bar 后调用
        iid: instrument id
        bar: bar

        """
        # print(bar)
        [handler(bar) for handler in self._bar_handlers[iid]]

    def _run_timer(self) -> None:
        """
        Sleep by interval second(s) and then generate a timer event.
        """
        while self._active:
            sleep(self._interval)
            event = Event(EVENT_TIMER)
            self.put(event)

    def _on_reqsubscribe(self):
        """有订阅需求时调用"""
        ps = self._rds.pubsub()
        ps.psubscribe(EVENT_REQ_SUBSCRIBE + '*')
        while self._active:
            '''data:type,pattern,channel,data'''
            for data in ps.listen():
                if data['type'] in ['message', 'pmessage']:
                    et = str(data['channel'][len(EVENT_REQ_SUBSCRIBE):])
                    data = json.loads(data['data'])
                    print(f'{et}, :, {len(et)}, :{type(data)}\n')

<<<<<<< HEAD
                    # self.ps.psubscribe(et + data)
                    # self._channels_queue.put(Event(et, data))
=======
>>>>>>> b767d7fb0750e8eb63cb0f7f8872c05ec5928302
                    pubsub = self._rds.pubsub()
                    pubsub.psubscribe(et + data)
                    if et==EVENT_REQ_SUBSCRIBE_TICK:
                        threading.Thread(target=self._tick_listen,args=(pubsub,) ).start()
                    elif et==EVENT_REQ_SUBSCRIBE_BAR:
                        threading.Thread(target=self._bar_listen, args=(pubsub,)).start()
                    else:
                        threading.Thread(target=self._listen, ).start()

    def req_subscribe(self, event: Event):
        self.publish(event_type=EVENT_REQ_SUBSCRIBE + event.event_type, data=event.data)

    def _req_subscribe_tick(self,iid):
        ps = self._rds.pubsub()
        ps.psubscribe(EVENT_REQ_SUBSCRIBE_TICK+iid)
        # print(self._tick_handlers)
        threading.Thread(target=self._tick_listen, args=(ps,)).start()

    def _req_subscribe_bar(self,iid):
        ps = self._rds.pubsub()
        ps.psubscribe(EVENT_REQ_SUBSCRIBE_BAR+iid)
        # print(self._bar_handlers)
        threading.Thread(target=self._bar_listen, args=(ps,) ).start()

    def put(self, event):
        self._queue.put(Event(event.event_type, event.data))

    def rds_put(self, event):
        self._rds.set(event.event_type, event.data)

    def start(self) -> None:
        """
        Start event engine to process events and generate timer events.
        """
        self._active = True
        self._thread.start()
        self._req_thread.start()
        # self._timer.start()

    def stop(self) -> None:
        """
        Stop event engine.
        """
        self._active = False
        self._timer.join()
        self._thread.join()
        self._req_thread.join()

    close = stop

    def publish(self, event: Event = None, event_type=None, data=None) -> None:
        """
        Publish an event object into redis server.
        """
        if event is None:
            self._rds.publish(event_type, json.dumps(data))
        else:
            # print(event.data)
            self._rds.publish(event.event_type, json.dumps(event.data))

    def register(self, event_type: str, handler: HandlerType) -> None:
        """
        Register a new handler function for a specific event type. Every
        function can only be registered once for each event type.
        """
        handler_list = self._handlers[event_type]
        if handler not in handler_list:            
            if EVENT_ON_TICK in event_type:
                iid=event_type.split('.')[1]
                self.register_tick(iid,handler)
            elif EVENT_ON_BAR in event_type:
                iid=event_type.split('.')[1]
                self.register_bar(iid,handler)
            else:            
                handler_list.append(handler)
                self.ps.subscribe(event_type)

    def register_tick(self, iid: str, handler: HandlerType) -> None:
        """
        Register a new handler function for a EVENT_ON_TICK event type. Every
        function can only be registered once for each event type.
        """
        handler_list = self._tick_handlers[iid]
        if not handler_list :
            handler_list.append(handler)
            self._req_subscribe_tick(iid)
        elif handler not in handler_list:
            handler_list.append(handler)


    def register_bar(self, iid: str, handler: HandlerType) -> None:
        """
        Register a new bar handler function for an iid. Every
        function can only be registered once for each iid.
        """
        handler_list = self._bar_handlers[iid]
        # print(handler_list)
        if not handler_list :
            handler_list.append(handler)
            self._req_subscribe_bar(iid)

        elif handler not in handler_list:
            handler_list.append(handler)

    def unregister(self, event_type: str, handler: HandlerType) -> None:
        """
        Unregister an existing handler function from event engine.
        """
        handler_list = self._handlers[event_type]

        if handler in handler_list:
            handler_list.remove(handler)

        if not handler_list:
            self._handlers.pop(event_type)
            self.ps.unsubscribe(event_type)

    def register_general(self, handler: HandlerType) -> None:
        """
        Register a new handler function for all event types. Every
        function can only be registered once for each event type.
        """
        if handler not in self._general_handlers:
            self._general_handlers.append(handler)

    def unregister_general(self, handler: HandlerType) -> None:
        """
        Unregister an existing general handler function.
        """
        if handler in self._general_handlers:
            self._general_handlers.remove(handler)


host = "120.79.67.37"
port = 6379
password = "steward1!"
pool = redis.ConnectionPool(host=host, port=port,
                            db=0, password=password, decode_responses=True)
rds = redis.StrictRedis(connection_pool=pool)
ree: RedisEventEngine = RedisEventEngine(rds=rds)

def RcvTick(tick):
    print(f"RcvTick:\n {tick}")
def RcvBar(bar):
    print(f"RcvBar:\n {bar}")

if __name__ == '__main__':
    # keys=ree._rds.keys()
    # print(keys)
    ree.start()
    print('subscribe')
    time.sleep(1)
    # ree.req_subscribe(Event('tick.', 'v2401'))
    # ree.req_subscribe(Event('tick.', 'TA401'))
    
    ree.register_tick('v2401',RcvTick)
    print('subscribe: v2401')

    # ree.stop()

